package org.bigdata.framework.core.elasticsearch;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

public class ElasticSearchService {

	private final static int MAX = 10000;

	TransportClient client;

	/**
	 * 功能描述：服务初始化
	 * 
	 * @param clusterName
	 *            集群名称
	 * @param ip
	 *            地址
	 * @param port
	 *            端口
	 */
	public ElasticSearchService(String clusterName, String ip, int port) {
		try {
			Settings settings = Settings.builder()
					.put("cluster.name", clusterName).build();

			this.client = new PreBuiltTransportClient(settings)
					.addTransportAddress(new InetSocketTransportAddress(
							InetAddress.getByName(ip), port));
		} catch (UnknownHostException e) {
			try {
				throw new Exception("es init failed!", e);
			} catch (Exception e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		}
	}

	/**
	 * 功能描述：新建索引
	 * 
	 * @param indexName
	 *            索引名
	 */
	public void createIndex(String indexName) {
		client.admin().indices().create(new CreateIndexRequest(indexName))
				.actionGet();
	}

	/**
	 * 功能描述：新建索引
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 */
	public void createIndex(String index, String type) {
		client.prepareIndex(index, type).setSource().get();
	}

	/**
	 * 功能描述：删除索引
	 * 
	 * @param index
	 *            索引名
	 * @throws Exception
	 */
	public void deleteIndex(String index) throws Exception {
		if (indexExist(index)) {
			DeleteIndexResponse dResponse = client.admin().indices()
					.prepareDelete(index).execute().actionGet();
			if (!dResponse.isAcknowledged()) {
				throw new Exception("failed to delete index.");
			}
		} else {
			throw new Exception("index name not exists");
		}
	}

	/**
	 * 功能描述：验证索引是否存在
	 * 
	 * @param index
	 *            索引名
	 */
	public boolean indexExist(String index) {
		IndicesExistsRequest inExistsRequest = new IndicesExistsRequest(index);
		IndicesExistsResponse inExistsResponse = client.admin().indices()
				.exists(inExistsRequest).actionGet();
		return inExistsResponse.isExists();
	}

	/**
	 * 功能描述：插入数据
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param json
	 *            数据
	 */
	public void insertData(String index, String type, String json) {
		IndexResponse response = client.prepareIndex(index, type)
				.setSource(json).get();
	}

	/**
	 * 功能描述：插入数据
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param _id
	 *            数据id
	 * @param json
	 *            数据
	 */
	public void insertData(String index, String type, String _id, String json) {
		IndexResponse response = client.prepareIndex(index, type).setId(_id)
				.setSource(json).get();
	}

	/**
	 * 功能描述：更新数据
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param _id
	 *            数据id
	 * @param json
	 *            数据
	 */
	public void updateData(String index, String type, String _id, String json)
			throws Exception {
		try {
			UpdateRequest updateRequest = new UpdateRequest(index, type, _id)
					.doc(json);
			client.update(updateRequest).get();
		} catch (Exception e) {
			throw new Exception("update data failed.", e);
		}
	}

	/**
	 * 功能描述：删除数据
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param _id
	 *            数据id
	 */
	public void deleteData(String index, String type, String _id) {
		DeleteResponse response = client.prepareDelete(index, type, _id).get();
	}

	/**
	 * 功能描述：批量插入数据
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param data
	 *            (_id 主键, json 数据)
	 */
	public void bulkInsertData(String index, String type,
			Map<String, String> data) {
		BulkRequestBuilder bulkRequest = client.prepareBulk();
		Iterator iter = data.entrySet().iterator();
		while (iter.hasNext()) {
			Map.Entry entry = (Map.Entry) iter.next();
			String field = entry.getKey().toString();
			Object value = entry.getValue();
			bulkRequest.add(client.prepareIndex(index, type, field).setSource(
					value));
		}

		BulkResponse bulkResponse = bulkRequest.get();
	}

	/**
	 * 功能描述：批量插入数据
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param jsonList
	 *            批量数据
	 */
	public void bulkInsertData(String index, String type, List<String> jsonList) {
		BulkRequestBuilder bulkRequest = client.prepareBulk();
		for (int i = 0; i < jsonList.size(); i++) {
			bulkRequest.add(client.prepareIndex(index, type).setSource(
					jsonList.get(i)));
		}
		BulkResponse bulkResponse = bulkRequest.get();
	}
	
	/**
	 * 查询总数
	 * @param index
	 * @param type
	 * @param constructor
	 * @return
	 */
	public Long searchCount(String index, String type,
			ESQueryBuilderConstructor constructor) {
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
				.setTypes(type);
		// 设置查询体
		searchRequestBuilder.setQuery(constructor.listBuilders());
		try {
			SearchResponse searchResponse = searchRequestBuilder.execute()
					.actionGet();
			SearchHits hits = searchResponse.getHits();
			return hits.getTotalHits();
		} catch (Exception e) {
			// TODO: handle exception
			return 0l;
		}
		
	}
	/**
	 * 
	* method: searchCountIsNotNull
	* Description:查询不为空的字段总数
	* @author 邢天龙
	* @date 2018年4月16日 下午5:57:37
	* @version 1.0
	 */
	public Long searchCountIsNotNull(String index, String type,
			ESQueryBuilderConstructor constructor,String fild) {
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
				.setTypes(type);
	    searchRequestBuilder.setPostFilter(QueryBuilders.existsQuery(fild));//过滤为空的值
		// 设置查询体
		searchRequestBuilder.setQuery(constructor.listBuilders());
		try {
			SearchResponse searchResponse = searchRequestBuilder.execute()
					.actionGet();
			SearchHits hits = searchResponse.getHits();
			
			return hits.getTotalHits();
		} catch (Exception e) {
			// TODO: handle exception
			return 0l;
		}
		
	}
	/**
	 * 
	* method: sortValue
	* Description：根据列名排序
	* @author 邢天龙
	* @date 2018年4月8日 下午6:14:56
	* @version 1.0
	 */
	public List<String> sortValue(String index, String type,
			ESQueryBuilderConstructor constructor,String field) {
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
		// 设置查询体 
		List<String> results = new ArrayList<>();
		searchRequestBuilder.setSize(constructor.getSize());
	     if(StringUtils.isNotEmpty(constructor.getDesc())){
	 		searchRequestBuilder.addSort(field+".keyword", SortOrder.DESC); 
	     }
	     if(StringUtils.isNotEmpty(constructor.getAsc())){
		   searchRequestBuilder.addSort(field+".keyword", SortOrder.ASC); 
		 }
		searchRequestBuilder.setQuery(constructor.listBuilders());
		try {
			SearchResponse searchResponse = searchRequestBuilder.execute()
					.actionGet();
			SearchHits hits = searchResponse.getHits();
			SearchHit[] searchHists = hits.getHits();
			for (SearchHit sh : searchHists) {
				results.add(sh.getSource().get(field).toString());
			}
		} catch (Exception e) {
			// TODO: handle exception
		}
		  return results;
	}
	/**
	 * 
	* method: getSumForColumnName
	* Description:
	* @author 邢天龙
	* @date 2018年4月13日 上午9:43:06
	* @version 1.0
	 */
	public List<Map<String,Object>> getSumForColumnName(String index, String type,ESQueryBuilderConstructor constructor, String groupByColumn, String sumColumn) {
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
		TermsAggregationBuilder groupBy=AggregationBuilders.terms(groupByColumn+".keyword").field(groupByColumn+".keyword");
		SumAggregationBuilder sum=AggregationBuilders.sum(sumColumn).field(sumColumn);
		TermsAggregationBuilder d=groupBy.subAggregation(sum);
		SearchResponse searchResponse=searchRequestBuilder.setQuery(constructor.listBuilders()).addAggregation(d).get();
		Terms stateAgg = searchResponse.getAggregations().get(groupByColumn+".keyword");
		JSONObject json=JSON.parseObject(searchResponse.toString());
		JSONObject aggregations=JSON.parseObject(json.get("aggregations").toString());
		JSONObject user_type=JSON.parseObject(aggregations.get(groupByColumn+".keyword").toString());
		List<Map<String,Object>> onj=(List<Map<String, Object>>) user_type.get("buckets");
		List<Map<String,Object>>list=new ArrayList();
		if(onj.size()>0){
			for(int i=0;i<onj.size();i++) {
			Map<String,Object> map=new HashMap();
			Map<String,Object>data=onj.get(i);
			String key=data.get("key").toString();
			Map resource_price=(Map) data.get(sumColumn);
			map.put("name", key);
			map.put("value", resource_price.get("value"));
			list.add(map);
			}
		 }
		return list;
	}
	/**
	 * 功能描述：查询
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param constructor
	 *            查询构造
	 */
	public List<Map<String, Object>> search(String index, String type,
			ESQueryBuilderConstructor constructor) {
		List<Map<String, Object>> result = new ArrayList<>();
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
				.setTypes(type);
		// 排序
		if (StringUtils.isNotEmpty(constructor.getAsc()))
			searchRequestBuilder.addSort(constructor.getAsc(), SortOrder.ASC);
		if (StringUtils.isNotEmpty(constructor.getDesc()))
			searchRequestBuilder.addSort(constructor.getDesc(), SortOrder.DESC);
		// 设置查询体
		searchRequestBuilder.setQuery(constructor.listBuilders());
		// 返回条目数
		int size = constructor.getSize();
		if (size < 0) {
			size = 0;
		}
		if (size > MAX) {
			size = MAX;
		}
		// 返回条目数
		searchRequestBuilder.setSize(size);

		searchRequestBuilder.setFrom(constructor.getFrom() < 0 ? 0
				: constructor.getFrom());
		
		try {
			SearchResponse searchResponse = searchRequestBuilder.execute()
					.actionGet();

			SearchHits hits = searchResponse.getHits();
			SearchHit[] searchHists = hits.getHits();
			for (SearchHit sh : searchHists) {
				result.add(sh.getSource());
			}
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
		
		return result;
	}

	/**
	 * 功能描述：统计查询
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param constructor
	 *            查询构造
	 * @param groupBy
	 *            统计字段
	 */
	public Map<Object, Object> statSearch(String index, String type,
			ESQueryBuilderConstructor constructor, String groupBy) {
		Map<Object, Object> map = new LinkedHashMap();
		SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
				.setTypes(type);
		// 排序
		if (StringUtils.isNotEmpty(constructor.getAsc()))
			searchRequestBuilder.addSort(constructor.getAsc(), SortOrder.ASC);
		if (StringUtils.isNotEmpty(constructor.getDesc()))
			searchRequestBuilder.addSort(constructor.getDesc(), SortOrder.DESC);
		// 设置查询体
		if (null != constructor) {
			searchRequestBuilder.setQuery(constructor.listBuilders());
		} else {
			searchRequestBuilder.setQuery(QueryBuilders.matchAllQuery());
		}
		int size = constructor.getSize();
		if (size < 0) {
			size = 0;
		}
		if (size > MAX) {
			size = MAX;
		}
		// 返回条目数
		searchRequestBuilder.setSize(size);

		searchRequestBuilder.setFrom(constructor.getFrom() < 0 ? 0
				: constructor.getFrom());
		SearchResponse sr = searchRequestBuilder.addAggregation(
				AggregationBuilders.terms("agg").field(groupBy)).get();

		Terms stateAgg = sr.getAggregations().get("agg");
		Iterator<Terms.Bucket> iter = (Iterator<Bucket>) stateAgg.getBuckets()
				.iterator();

		while (iter.hasNext()) {
			Terms.Bucket gradeBucket = iter.next();
			map.put(gradeBucket.getKey(), gradeBucket.getDocCount());
		}

		return map;
	}

	/**
	 * 功能描述：统计查询
	 * 
	 * @param index
	 *            索引名
	 * @param type
	 *            类型
	 * @param constructor
	 *            查询构造
	 * @param agg
	 *            自定义计算
	 */
	public Map<String, Object> statSearch(String index, String type,
			ESQueryBuilderConstructor constructor, AggregationBuilder agg,String condition) {
		 SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
		if (agg == null) {
			return null;
		}
		Map<String, Object> map = new LinkedHashMap<String, Object>();
		try {
			SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
					.setTypes(type);
			// 排序
			if (StringUtils.isNotEmpty(constructor.getAsc()))
				searchRequestBuilder.addSort(constructor.getAsc(), SortOrder.ASC);
			if (StringUtils.isNotEmpty(constructor.getDesc()))
				searchRequestBuilder.addSort(constructor.getDesc(), SortOrder.DESC);
			// 设置查询体
			if (null != constructor) {
				searchRequestBuilder.setQuery(constructor.listBuilders());
			} else {
				searchRequestBuilder.setQuery(QueryBuilders.matchAllQuery());
			}
			int size = constructor.getSize();
			if (size < 0) {
				size = 0;
			}
			if (size > MAX) {
				size = MAX;
			}
			// 返回条目数
			searchRequestBuilder.setSize(size);

			searchRequestBuilder.setFrom(constructor.getFrom() < 0 ? 0
					: constructor.getFrom());
			SearchResponse sr = searchRequestBuilder.addAggregation(agg).get();

			Terms stateAgg = sr.getAggregations().get("agg");
			Iterator<Terms.Bucket> iter = (Iterator<Bucket>) stateAgg.getBuckets()
					.iterator();
			while (iter.hasNext()) {
				Terms.Bucket gradeBucket = iter.next();
				if("dateformat".equals(condition)){//根据业务需要格式化日期
				    map.put(df.format(gradeBucket.getKey()), gradeBucket.getDocCount());
				}else{
				   map.put(gradeBucket.getKey().toString(), gradeBucket.getDocCount());	
				}
			}
		} catch (Exception e) {
			// TODO: handle exception
		}

		return map;
	}
	
	public Map<Object, Object> statSearch(String index, String type,
			ESQueryBuilderConstructor constructor, AggregationBuilder agg) {
		 SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
		if (agg == null) {
			return null;
		}
		Map<Object, Object> map = new LinkedHashMap<Object, Object>();
		try {
			SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
					.setTypes(type);
			// 排序
			if (StringUtils.isNotEmpty(constructor.getAsc()))
				searchRequestBuilder.addSort(constructor.getAsc(), SortOrder.ASC);
			if (StringUtils.isNotEmpty(constructor.getDesc()))
				searchRequestBuilder.addSort(constructor.getDesc(), SortOrder.DESC);
			// 设置查询体
			if (null != constructor) {
				searchRequestBuilder.setQuery(constructor.listBuilders());
			} else {
				searchRequestBuilder.setQuery(QueryBuilders.matchAllQuery());
			}
			int size = constructor.getSize();
			if (size < 0) {
				size = 0;
			}
			if (size > MAX) {
				size = MAX;
			}
			// 返回条目数
			searchRequestBuilder.setSize(size);

			searchRequestBuilder.setFrom(constructor.getFrom() < 0 ? 0
					: constructor.getFrom());
			SearchResponse sr = searchRequestBuilder.addAggregation(agg).get();

			Terms stateAgg = sr.getAggregations().get("agg");
			Iterator<Terms.Bucket> iter = (Iterator<Bucket>) stateAgg.getBuckets()
					.iterator();
			while (iter.hasNext()) {
				Terms.Bucket gradeBucket = iter.next();
				map.put(gradeBucket.getKey().toString(), gradeBucket.getDocCount());	
			}
		} catch (Exception e) {
			// TODO: handle exception
		}

		return map;
	}
	/**
	 * 功能描述：关闭链接
	 */
	public void close() {
		client.close();
	}

}
